package com.zjl.Kafka.第03章_操作kafka;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.junit.Test;

import java.time.Duration;
import java.util.*;

/**
 * 消费者组偏移量文件  __consumer_offsets-xx  xx -> 0-49
 *
 * 消费者组消费的偏移量
 *     默认偏移量为 LEO (每个节点在指定的主题下的最大偏移量)，这样的话，消费者启动晚了，就永远消费不到之前的了
 *     earliest：最小偏移量
 *
 *
 *
 *
 */
public class E1_消费者获取数据 {
    public static void main(String[] args) {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "82.157.71.243:9092");//连接地址
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());//key 的 反序列化
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());//value 的 反序列化

        //提交了才可以读取  默认  read_uncommitted：读未提交       read_committed：读已提交
        config.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed");

        //消费者组的名称,  kafka 会保存消费者组的偏移量，他优先级最高比其他偏移量高
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "x01");//必选项
        //__consumer_offsets-xx    消费者偏移量文件
        config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);//是否自动提交偏移量  默认 true 自动提交
//        config.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);//自动提交时间  默认5000 = 5秒


        // 默认偏移量为 LEO (每个节点在指定的主题下的最大偏移量)，这样的话，消费者启动晚了，就永远消费不到之前的了
        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");//最小偏移量
        //创建消费者对象
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(config);

        //订阅主题
        List<String> topics = new ArrayList<>();
        topics.add("x01");
        topics.add("x03");
        kafkaConsumer.subscribe(topics);

        //指定偏移量消费
        kafkaConsumer.poll(Duration.ofMillis(500));//从生产者拉取数据
        Set<TopicPartition> assignments = kafkaConsumer.assignment();//获取每个主题的信息
        for (TopicPartition topicPartition:assignments){
            if("x01".equals(topicPartition.topic())){//是指定主题
                kafkaConsumer.seek(topicPartition,2);//修改主题偏移量
            }
        }


        while (true) {
            //从生产者拉取数据
            ConsumerRecords<String, String> polls = kafkaConsumer.poll(Duration.ofMillis(500));

            System.out.println(polls.partitions().size());

            for (TopicPartition tops : polls.partitions()) {

                List<ConsumerRecord<String, String>> records = polls.records(tops);
                for (ConsumerRecord<String, String> pool : records) {

                    System.out.println(pool);
                }
            }
//            kafkaConsumer.commitAsync();//手动保存偏移量  异步提交  优选
            kafkaConsumer.commitSync();//手动保存偏移量   同步提交 同步完才继续操作
        }

        //关闭消费者
        //kafkaConsumer.close();

    }
}
